package com.danan.realtime.util;

import com.alibaba.fastjson.JSONObject;
import com.danan.realtime.pojo.TableProcess;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @author NanHuang
 * @Date 2023/1/24
 */
public class MyPhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {

    private Connection phoenixConn;
    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        phoenixConn = JDBCUtil.getPhoenixConnection();
        jedis = RedisUtil.getJedis();
    }

    @Override
    public void close() throws Exception {
        phoenixConn.close();
        jedis.close();
    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> value, Context context) throws Exception {
        JSONObject data = value.f0;
        TableProcess config = value.f1;
        // 1 将数据写入Phoenix
        insertDataToPhoenix(data, config);
        // 2 删除Redis中的旧数据
        jedis.del(config.getSinkTable() + ":" + data.getString("id"));
    }

    private void insertDataToPhoenix(JSONObject data, TableProcess config) throws SQLException {
        // 1 拼接导数SQL
        String template = "upsert into %s values (%s)";
        String columns = config.getSinkColumns();
        String sql = String.format(template, config.getSinkTable(), columns.replaceAll("[^,]+", "?"));
        // 2 预编译
        PreparedStatement ps = phoenixConn.prepareStatement(sql);
        // 3 占位符赋值
        String[] columnWords = columns.split(",");
        for (int i = 0; i < columnWords.length; i++) {
            String columnData = data.getString(columnWords[i]);
            ps.setString(i + 1,columnData);
        }
        // 4 运行SQL
        ps.execute();
        // 5 提交事务
        phoenixConn.commit();
        // 6 关闭预编译
        ps.close();
    }
}
